package com.example.sangong.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

@Component
@Slf4j
public class KafkaSendHelper {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMsgAsync(String topic, String message) {
        ListenableFuture<SendResult<String, String>> feature = kafkaTemplate.send(topic, message);
        feature.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("kafka向{}主题发送消息'{}'异常了", topic, message);
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                ProducerRecord<String, String> producerRecord = result.getProducerRecord();
                log.info("kafka向{}主题发送消息'{}'成功了", producerRecord.topic(), producerRecord.value());
            }
        });
    }
}
